package com.jfireframework.commonloggersolution;

import com.jfireframework.baseutil.StringUtil;
import com.jfireframework.baseutil.TRACEID;
import com.jfireframework.baseutil.reflect.ReflectUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public abstract class CommonFileLoggerImpl<V>
{
    public static final    int              RECORDS_FILE_SIZE;
    public static final    long             SHIFT;
    protected static final Logger           LOGGER             = LoggerFactory.getLogger(CommonFileLoggerImpl.class);
    protected              MappedByteBuffer metadataWritePosi;
    protected              MappedByteBuffer metadataReadPosi;
    protected              MappedByteBuffer recordsBuffer;
    protected              MappedByteBuffer snapshotBuffer;
    protected              CycleBuffer      writeBuffer;
    protected              CycleBuffer      readBuffer;
    protected volatile     long             readPosi           = 0;
    protected volatile     long             writePosi          = 0;
    protected              int              writeCount;
    protected              Charset          charset            = Charset.forName("utf8");
    protected              WriteLock        writeLock          = new WriteLock();
    protected final        float            SNAPSHOT_FACTOR;
    protected final        int              THRESHOLD;
    ///
    protected              ExecutorService  snapshotWorker     = Executors.newSingleThreadExecutor();
    protected              AtomicInteger    snapshotWorkerFlag = new AtomicInteger(IDLE);
    private static final   int              IDLE               = 0;
    private static final   int              WORK               = 1;

    static
    {
        String property = System.getProperty("common.logger.solution.filesize");
        if (StringUtil.isNotBlank(property))
        {
            RECORDS_FILE_SIZE = Integer.parseInt(property);
            SHIFT = RECORDS_FILE_SIZE - 1;
        }
        else
        {
            RECORDS_FILE_SIZE = 1024 * 1024 * 8;
            SHIFT = RECORDS_FILE_SIZE - 1;
        }
    }

    public CommonFileLoggerImpl(File dict)
    {
        this(dict, 0.75f);
    }

    public CommonFileLoggerImpl(File dict, float factor)
    {
        SNAPSHOT_FACTOR = factor;
        THRESHOLD = (int) (RECORDS_FILE_SIZE * (1 - factor));
        String traceId = TRACEID.currentTraceId();
        if (dict.exists() == false)
        {
            dict.mkdirs();
        }
        if (dict.isDirectory() == false)
        {
            throw new IllegalArgumentException(dict.getAbsolutePath() + "不是一个文件夹路径");
        }
        try
        {
            LOGGER.debug("traceId:{} 日志文件存放路径为:{}", traceId, dict.getAbsoluteFile());
            File metadata = new File(dict, "metadata");
            if (metadata.exists() == false)
            {
                metadata.createNewFile();
                RandomAccessFile randomAccessFile = new RandomAccessFile(metadata, "rw");
                randomAccessFile.setLength(16);
                randomAccessFile.close();
            }
            metadataReadPosi = new RandomAccessFile(metadata, "rw").getChannel().map(FileChannel.MapMode.READ_WRITE, 0, 8);
            metadataWritePosi = new RandomAccessFile(metadata, "rw").getChannel().map(FileChannel.MapMode.READ_WRITE, 8, 8);
            File records = new File(dict, "records");
            if (records.exists() == false)
            {
                records.createNewFile();
                RandomAccessFile randomAccessFile = new RandomAccessFile(records, "rw");
                randomAccessFile.setLength(RECORDS_FILE_SIZE);
                randomAccessFile.close();
            }
            recordsBuffer = new RandomAccessFile(records, "rw").getChannel().map(FileChannel.MapMode.READ_WRITE, 0, RECORDS_FILE_SIZE);
            File snapshot = new File(dict, "snapshot");
            if (snapshot.exists() == false)
            {
                snapshot.createNewFile();
                RandomAccessFile randomAccessFile = new RandomAccessFile(snapshot, "rw");
                randomAccessFile.setLength(RECORDS_FILE_SIZE);
                randomAccessFile.close();
            }
            snapshotBuffer = new RandomAccessFile(snapshot, "rw").getChannel().map(FileChannel.MapMode.READ_WRITE, 0, RECORDS_FILE_SIZE);
            readPosi = metadataReadPosi.getLong();
            writePosi = metadataWritePosi.getLong();
            LOGGER.debug("traceId:{} 元数据文件读取显示，逻辑读取偏移量:{},逻辑写入偏移量:{}", traceId, readPosi, writePosi);
            generateWriteBuffer();
            generateReadBuffer();
        }
        catch (IOException e)
        {
            ReflectUtil.throwException(e);
        }
    }

    private void generateReadBuffer()
    {
        int        physicsReadPosi  = (int) (readPosi & SHIFT);
        int        physicsWritePosi = (int) (writePosi & SHIFT);
        ByteBuffer slice            = recordsBuffer.duplicate();
        readBuffer = new CycleBuffer(slice, physicsReadPosi, physicsWritePosi, BufferMode.read);
    }

    private void generateWriteBuffer()
    {
        long end              = readPosi + SHIFT;
        int  physicsWritePosi = (int) (writePosi & SHIFT);
        int  physicsEnd       = (int) (end & SHIFT);
        writeBuffer = new CycleBuffer(recordsBuffer, physicsWritePosi, physicsEnd, BufferMode.write);
    }

    private MappedByteBuffer mapFile(File metadata, int posi, int size) throws IOException
    {
        RandomAccessFile randomAccessFile = new RandomAccessFile(metadata, "rw");
        randomAccessFile.setLength(size);
        return randomAccessFile.getChannel().map(FileChannel.MapMode.READ_WRITE, posi, size);
    }

    protected final void startRecord()
    {
        if (writeLock.lock() == false)
        {
            throw new IllegalStateException("该方法的调用方应该本身实现了互斥，比如Sync关键字，需要避免在数据记录时进行争用");
        }
        long lsn = writePosi;
        checkRemain(8);
        writeBuffer.putLong(lsn);
        writeCount = 8;
        LOGGER.debug("traceId:{} 准备开始写入新纪录，当前写入位置:{}", TRACEID.currentTraceId(), writePosi);
    }

    protected final void endRecord()
    {
        long newWritePosi = writePosi + writeCount;
        metadataWritePosi.putLong(0, newWritePosi);
        writeCount = 0;
        writePosi = newWritePosi;
        LOGGER.debug("traceId:{} 数据写入完毕，新的写入位置:{}", TRACEID.currentTraceId(), writePosi);
        writeLock.release();
    }

    protected final void write(byte[] value)
    {
        checkRemain(value.length);
        writeBuffer.put(value);
        writeCount += value.length;
    }

    protected final void write(byte value)
    {
        checkRemain(1);
        writeBuffer.put(value);
        writeCount++;
    }

    protected final void writeInt(int value)
    {
        checkRemain(4);
        writeBuffer.putInt(value);
        writeCount += 4;
    }

    protected final void writeShort(short value)
    {
        checkRemain(2);
        writeBuffer.putShort(value);
        writeCount += 2;
    }

    protected final void writeLong(long value)
    {
        checkRemain(8);
        writeBuffer.putLong(value);
        writeCount += 8;
    }

    private void checkRemain(int length)
    {
        if (writeLock.check() == false)
        {
            throw new IllegalStateException("当前方法线程没有持有写入锁");
        }
        final int remaining = writeBuffer.remaining();
        if (remaining < THRESHOLD)
        {
            startSnapshotWorker(remaining);
        }
        if (remaining < length)
        {
            long end = readPosi + SHIFT;
            LOGGER.debug("traceId:{} 剩余可写入空间不足，触发强制快照回收空间.当前写入起始位置:{},可写终止位置:{}", TRACEID.currentTraceId(), writePosi, end);
            //执行快照流程，将空间回收
            snapshot();
            end = readPosi + SHIFT;
            LOGGER.debug("traceId:{} 强制快照回收后，当前可写终止位置:{}", TRACEID.currentTraceId(), end);
            if (writeBuffer.remaining() < length)
            {
                throw new IllegalStateException("无法回收到足够的空间进行写入");
            }
        }
    }

    private void startSnapshotWorker(final int remaining)
    {
        if (snapshotWorkerFlag.get() == IDLE && snapshotWorkerFlag.compareAndSet(IDLE, WORK))
        {
            snapshotWorker.submit(new Runnable()
            {
                @Override
                public void run()
                {
                    String traceId = TRACEID.newTraceId();
                    LOGGER.debug("traceId:{} 当前剩余写入空间:{}，低于阀值:{},启动快照任务", traceId, remaining, THRESHOLD);
                    try
                    {
                        snapshot();
                    }
                    finally
                    {
                        snapshotWorkerFlag.set(IDLE);
                    }
                }
            });
        }
    }

    /**
     * 执行快照操作。在快照完毕后，返回日志的当前数据。可能存在三种情况：<br/>
     * 1. 快照文件非法，日志文件没有内容，返回null。
     * 2. 快照文件合法，日志文件没有内容，返回null。
     * 3. 快照文件合法，日志文件有内容，合并快照和日志文件后，返回当前的业务数据
     *
     * @return
     */
    protected final synchronized V snapshot()
    {
        String traceId = TRACEID.currentTraceId();
        LOGGER.debug("traceId:{} 准备进行快照", traceId);
        Snapshot<V> snapshot       = readExistSnapshot();
        long        snapshotLsn    = snapshot.legal ? snapshot.lsn : 0;
        long        localReadPosi  = readPosi > snapshotLsn ? readPosi : snapshotLsn;
        long        localWritePosi = writePosi;
        if (localReadPosi == localWritePosi)
        {
            LOGGER.debug("traceId:{} 日志文件的逻辑读取和写入位置均为:{},当前不存在需要读取的内容", traceId, localReadPosi);
            if (snapshot.legal)
            {
                return snapshot.data;
            }
            else
            {
                return null;
            }
        }
        else
        {
            LOGGER.debug("traceId:{} 日志文件的逻辑读取位置是:{},逻辑写入位置是:{},需要读取字节数:{}", traceId, localReadPosi, localWritePosi, (localWritePosi - localReadPosi));
            long lsn              = localWritePosi;
            int  physicsReadPosi  = (int) (localReadPosi & SHIFT);
            int  physicsWritePosi = (int) (localWritePosi & SHIFT);
            readBuffer.resetStartAndEnd(physicsReadPosi, physicsWritePosi);
            V      archive = archive(snapshot, readBuffer);
            byte[] content = buildSnapshotContentFromArchive(archive);
            short  crc16   = Crc16.calcCrc16(content);
            writeToSnapshotBuffer(lsn, content, crc16);
            metadataReadPosi.putLong(0, lsn);
            readPosi = lsn;
            LOGGER.debug("traceId:{} 快照成功，读取偏移量更新为:{}", traceId, readPosi);
            int end   = (int) ((readPosi + SHIFT) & SHIFT);
            int start = (int) (localWritePosi & SHIFT);
            writeBuffer.resetStartAndEnd(start, end);
            return archive;
        }
    }

    private void writeToSnapshotBuffer(long lsn, byte[] content, short crc16)
    {
        String traceId = TRACEID.currentTraceId();
        LOGGER.debug("traceId:{} 准备写入快照文件，lsn:{},crc16:{},len:{}", traceId, lsn, crc16, content.length);
        snapshotBuffer.clear();
        snapshotBuffer.putLong(lsn);
        snapshotBuffer.putShort(crc16);
        snapshotBuffer.putInt(content.length);
        snapshotBuffer.put(content);
    }

    private Snapshot<V> readExistSnapshot()
    {
        String traceId = TRACEID.currentTraceId();
        snapshotBuffer.position(0);
        long lsn = snapshotBuffer.getLong();
        if (lsn == 0)
        {
            LOGGER.debug("traceId:{} 快照文件尚未记录数据，直接返回", traceId);
            Snapshot snapshot = new Snapshot();
            snapshot.legal = false;
            return snapshot;
        }
        short crc16 = snapshotBuffer.getShort();
        int   len   = snapshotBuffer.getInt();
        snapshotBuffer.mark();
        snapshotBuffer.limit(snapshotBuffer.position() + len);
        short calcCrc16 = Crc16.calcCrc16(snapshotBuffer);
        snapshotBuffer.reset();
        if (calcCrc16 != crc16)
        {
            LOGGER.debug("traceId:{} 快照文件crc校验失败", traceId);
            Snapshot snapshot = new Snapshot();
            snapshot.legal = false;
            return snapshot;
        }
        LOGGER.debug("traceId:{} 快照文件crc校验成功，准备读取快照内容", traceId);
        V        data     = doReadSnapshotBuffer(snapshotBuffer);
        Snapshot snapshot = new Snapshot();
        snapshot.lsn = lsn;
        snapshot.legal = true;
        snapshot.data = data;
        return snapshot;
    }

    /**
     * 根据输入的快照文件内容和记录日志文件，进行归档操作，并且返回归档后的业务数据。
     *
     * @param snapshot
     * @param buffer
     * @return
     */
    protected abstract V archive(Snapshot<V> snapshot, CycleBuffer buffer);

    /**
     * 读取快照文件内容，返回业务数据
     *
     * @param snapshotBuffer
     * @return
     */
    protected abstract V doReadSnapshotBuffer(ByteBuffer snapshotBuffer);

    /**
     * 以输入的业务归档数据，生成快照内容
     *
     * @param archive
     * @return
     */
    protected abstract byte[] buildSnapshotContentFromArchive(V archive);
}
